/*-------------------------------------------------------------------------
 *
 * libpqwalreceiver.c
 *
 * 此文件包含 walreceiver 的 libpq 特定部分。
 * 它作为动态模块加载，以避免将主服务器二进制文件与
 * libpq 链接。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>
#include <sys/time.h>

#include "access/xlog.h"
#include "catalog/pg_type.h"
#include "common/connect.h"
#include "funcapi.h"
#include "libpq-fe.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "pqexpbuffer.h"
#include "replication/walreceiver.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/tuplestore.h"
#ifdef FDDM
#include "replication/worker_internal.h"
#endif //FDDM

PG_MODULE_MAGIC;

void		_PG_init(void);

struct WalReceiverConn
{
	/* 当前与主服务器的连接（如有） */
	PGconn	   *streamConn;
	/* 用于记住连接是逻辑的还是物理的 */
	bool		logical;
	/* 当前读取记录的缓冲区 */
	char	   *recvBuf;
};

/* 接口函数的原型 */
static WalReceiverConn *fc_libpqrcv_connect(const char *fc_conninfo,
										 bool fc_logical, const char *fc_appname,
										 char **fc_err);
static void fc_libpqrcv_check_conninfo(const char *fc_conninfo);
static char *fc_libpqrcv_get_conninfo(WalReceiverConn *fc_conn);
static void fc_libpqrcv_get_senderinfo(WalReceiverConn *fc_conn,
									char **fc_sender_host, int *fc_sender_port);
static char *fc_libpqrcv_identify_system(WalReceiverConn *fc_conn,
									  TimeLineID *fc_primary_tli);
static int	fc_libpqrcv_server_version(WalReceiverConn *fc_conn);
static void fc_libpqrcv_readtimelinehistoryfile(WalReceiverConn *fc_conn,
											 TimeLineID fc_tli, char **fc_filename,
											 char **fc_content, int *fc_len);
static bool fc_libpqrcv_startstreaming(WalReceiverConn *fc_conn,
									const WalRcvStreamOptions *fc_options);
static void fc_libpqrcv_endstreaming(WalReceiverConn *fc_conn,
								  TimeLineID *fc_next_tli);
static int	fc_libpqrcv_receive(WalReceiverConn *fc_conn, char **fc_buffer,
							 pgsocket *fc_wait_fd);
static void fc_libpqrcv_send(WalReceiverConn *fc_conn, const char *fc_buffer,
						  int fc_nbytes);
#ifdef FDD
static char *fc_libpqrcv_create_slot(WalReceiverConn *fc_conn,
								  const char *fc_slotname,
								  bool fc_temporary,
								  bool fc_two_phase,
								  bool fc_failover,
								  CRSSnapshotAction fc_snapshot_action,
								  XLogRecPtr *fc_lsn);
#else
static char *libpqrcv_create_slot(WalReceiverConn *conn,
								  const char *slotname,
								  bool temporary,
								  bool two_phase,
								  CRSSnapshotAction snapshot_action,
								  XLogRecPtr *lsn);
#endif
static pid_t fc_libpqrcv_get_backend_pid(WalReceiverConn *fc_conn);
static WalRcvExecResult *fc_libpqrcv_exec(WalReceiverConn *fc_conn,
									   const char *fc_query,
									   const int fc_nRetTypes,
									   const Oid *fc_retTypes);
static void fc_libpqrcv_disconnect(WalReceiverConn *fc_conn);

static WalReceiverFunctionsType PQWalReceiverFunctions = {
	fc_libpqrcv_connect,
	fc_libpqrcv_check_conninfo,
	fc_libpqrcv_get_conninfo,
	fc_libpqrcv_get_senderinfo,
	fc_libpqrcv_identify_system,
	fc_libpqrcv_server_version,
	fc_libpqrcv_readtimelinehistoryfile,
	fc_libpqrcv_startstreaming,
	fc_libpqrcv_endstreaming,
	fc_libpqrcv_receive,
	fc_libpqrcv_send,
	fc_libpqrcv_create_slot,
	fc_libpqrcv_get_backend_pid,
	fc_libpqrcv_exec,
	fc_libpqrcv_disconnect
};

/* 私有函数的原型 */
static PGresult *fc_libpqrcv_PQexec(PGconn *fc_streamConn, const char *fc_query);
static PGresult *fc_libpqrcv_PQgetResult(PGconn *fc_streamConn);
static char *fc_stringlist_to_identifierstr(PGconn *fc_conn, List *fc_strings);

/*
 * 模块初始化函数
 */
void _PG_init(void)
{
	if (WalReceiverFunctions != NULL)
		elog(ERROR, "libpqwalreceiver already loaded");
	WalReceiverFunctions = &PQWalReceiverFunctions;
}

/*
 * 建立与主服务器的连接以进行 XLOG 流式传输
 *
 * 出错时返回 NULL，并用 palloc' 的错误消息填充 err。
 */
static WalReceiverConn *
fc_libpqrcv_connect(const char *fc_conninfo, bool fc_logical, const char *fc_appname,
				 char **fc_err)
{
	WalReceiverConn *fc_conn;
	PostgresPollingStatusType fc_status;
	const char *fc_keys[6];
	const char *fc_vals[6];
	int			fc_i = 0;

	/*
	 * 我们使用 expand_dbname 参数处理连接字符串（或
	 * URI），并传递一些额外的选项。
	 */
	fc_keys[fc_i] = "dbname";
	fc_vals[fc_i] = fc_conninfo;
	fc_keys[++fc_i] = "replication";
	fc_vals[fc_i] = fc_logical ? "database" : "true";
	if (!fc_logical)
	{
		/*
		 * 数据库名称在复制模式中被服务器忽略，但
		 * 指定“复制”以进行 .pgpass 查找。
		 */
		fc_keys[++fc_i] = "dbname";
		fc_vals[fc_i] = "replication";
	}
	fc_keys[++fc_i] = "fallback_application_name";
	fc_vals[fc_i] = fc_appname;
	if (fc_logical)
	{
		/* 告诉发布者转换为我们的编码 */
		fc_keys[++fc_i] = "client_encoding";
		fc_vals[fc_i] = GetDatabaseEncodingName();

		/*
		 * 强制各种 GUC 参数设置，以确保发布者将以
		 * 对订阅者明确的数据值形式输出。 （我们不想修改订阅者的 GUC
		 * 设置，因为这可能会让在
		 * 订阅者中运行的用户定义代码（如触发器）感到惊讶。）这应该与 pg_dump
		 * 的行为相匹配。
		 */
		fc_keys[++fc_i] = "options";
		fc_vals[fc_i] = "-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
	}
	fc_keys[++fc_i] = NULL;
	fc_vals[fc_i] = NULL;

	Assert(fc_i < sizeof(fc_keys));

	fc_conn = palloc0(sizeof(WalReceiverConn));
	fc_conn->streamConn = PQconnectStartParams(fc_keys, fc_vals,
											 /* expand_dbname = */ true);
	if (PQstatus(fc_conn->streamConn) == CONNECTION_BAD)
		goto bad_connection_errmsg;

	/*
	 * 循环连接，直到我们获得 OK 或 FAILED 状态。
	 *
	 * 根据 PQconnectPoll 的规格，首先等待套接字可写就绪。
	 */
	fc_status = PGRES_POLLING_WRITING;
	do
	{
		int			fc_io_flag;
		int			fc_rc;

		if (fc_status == PGRES_POLLING_READING)
			fc_io_flag = WL_SOCKET_READABLE;
#ifdef WIN32
		/* Windows 在等待连接建立时需要不同的测试 */
		else if (PQstatus(conn->streamConn) == CONNECTION_STARTED)
			fc_io_flag = WL_SOCKET_CONNECTED;
#endif
		else
			fc_io_flag = WL_SOCKET_WRITEABLE;

		fc_rc = WaitLatchOrSocket(MyLatch,
							   WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | fc_io_flag,
							   PQsocket(fc_conn->streamConn),
							   0,
							   WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);

		/* 中断了吗？ */
		if (fc_rc & WL_LATCH_SET)
		{
			ResetLatch(MyLatch);
			ProcessWalRcvInterrupts();
		}

		/* 如果套接字准备好，推进 libpq 状态机 */
		if (fc_rc & fc_io_flag)
			fc_status = PQconnectPoll(fc_conn->streamConn);
	} while (fc_status != PGRES_POLLING_OK && fc_status != PGRES_POLLING_FAILED);

	if (PQstatus(fc_conn->streamConn) != CONNECTION_OK)
		goto bad_connection_errmsg;

	if (fc_logical)
	{
		PGresult   *fc_res;

		fc_res = fc_libpqrcv_PQexec(fc_conn->streamConn,
							  ALWAYS_SECURE_SEARCH_PATH_SQL);
		if (PQresultStatus(fc_res) != PGRES_TUPLES_OK)
		{
			PQclear(fc_res);
			*fc_err = psprintf(_("could not clear search path: %s"),
							pchomp(PQerrorMessage(fc_conn->streamConn)));
			goto bad_connection;
		}
		PQclear(fc_res);
	}

	fc_conn->logical = fc_logical;

	return fc_conn;

	/* 错误路径，使用 libpq 的错误消息 */
bad_connection_errmsg:
	*fc_err = pchomp(PQerrorMessage(fc_conn->streamConn));

	/* 错误路径，错误已设置 */
bad_connection:
	PQfinish(fc_conn->streamConn);
	pfree(fc_conn);
	return NULL;
}

/*
 * 验证连接信息字符串（只尝试解析它）
 */
static void fc_libpqrcv_check_conninfo(const char *fc_conninfo)
{
	PQconninfoOption *fc_opts = NULL;
	char	   *fc_err = NULL;

	fc_opts = PQconninfoParse(fc_conninfo, &fc_err);
	if (fc_opts == NULL)
	{
		/* 错误字符串是 malloc' 的，因此我们必须显式释放它 */
		char	   *fc_errcopy = fc_err ? pstrdup(fc_err) : "out of memory";

		PQfreemem(fc_err);
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("invalid connection string syntax: %s", fc_errcopy)));
	}

	PQconninfoFree(fc_opts);
}

/*
 * 返回用户可显示的 conninfo 字符串。任何安全敏感字段
 * 都是经过模糊处理的。
 */
static char * fc_libpqrcv_get_conninfo(WalReceiverConn *fc_conn)
{
	PQconninfoOption *fc_conn_opts;
	PQconninfoOption *fc_conn_opt;
	PQExpBufferData fc_buf;
	char	   *fc_retval;

	Assert(fc_conn->streamConn != NULL);

	initPQExpBuffer(&fc_buf);
	fc_conn_opts = PQconninfo(fc_conn->streamConn);

	if (fc_conn_opts == NULL)
#ifdef FDD //cppcheck
	{
#endif
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("could not parse connection string: %s",
						_("out of memory"))));
#ifdef FDD //cppcheck
		return NULL;
	}
#endif
	/* 从各部分构建一个干净的连接字符串 */
	for (fc_conn_opt = fc_conn_opts; fc_conn_opt->keyword != NULL; fc_conn_opt++)
	{
		bool		fc_obfuscate;

		/* 跳过调试和空选项 */
		if (strchr(fc_conn_opt->dispchar, 'D') ||
			fc_conn_opt->val == NULL ||
			fc_conn_opt->val[0] == '\0')
			continue;

		/* 混淆安全敏感选项 */
		fc_obfuscate = strchr(fc_conn_opt->dispchar, '*') != NULL;

		appendPQExpBuffer(&fc_buf, "%s%s=%s",
						  fc_buf.len == 0 ? "" : " ",
						  fc_conn_opt->keyword,
						  fc_obfuscate ? "********" : fc_conn_opt->val);
	}

	PQconninfoFree(fc_conn_opts);

	fc_retval = PQExpBufferDataBroken(fc_buf) ? NULL : pstrdup(fc_buf.data);
	termPQExpBuffer(&fc_buf);
	return fc_retval;
}

/*
 * 提供此 WAL 接收器连接到的发送方的信息。
 */
static void fc_libpqrcv_get_senderinfo(WalReceiverConn *fc_conn, char **fc_sender_host,
						int *fc_sender_port)
{
	char	   *fc_ret = NULL;

	*fc_sender_host = NULL;
	*fc_sender_port = 0;

	Assert(fc_conn->streamConn != NULL);

	fc_ret = PQhost(fc_conn->streamConn);
	if (fc_ret && strlen(fc_ret) != 0)
		*fc_sender_host = pstrdup(fc_ret);

	fc_ret = PQport(fc_conn->streamConn);
	if (fc_ret && strlen(fc_ret) != 0)
		*fc_sender_port = atoi(fc_ret);
}

/*
 * 检查主系统标识符是否与我们的匹配，并获取当前主的时间线 ID。
 */
static char * fc_libpqrcv_identify_system(WalReceiverConn *fc_conn, TimeLineID *fc_primary_tli)
{
	PGresult   *fc_res;
	char	   *fc_primary_sysid;

	/*
	 * 从主服务器获取系统标识符和时间线 ID，作为 DataRow 消息。
	 */
	fc_res = fc_libpqrcv_PQexec(fc_conn->streamConn, "IDENTIFY_SYSTEM");
	if (PQresultStatus(fc_res) != PGRES_TUPLES_OK)
	{
		PQclear(fc_res);
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("could not receive database system identifier and timeline ID from "
						"the primary server: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));
	}
	/*
	 * IDENTIFY_SYSTEM 在 9.3 及更早版本返回 3 列，在 9.4 及以后返回 4 列。
	 */
	if (PQnfields(fc_res) < 3 || PQntuples(fc_res) != 1)
	{
		int			fc_ntuples = PQntuples(fc_res);
		int			fc_nfields = PQnfields(fc_res);

		PQclear(fc_res);
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("invalid response from primary server"),
				 errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
						   fc_ntuples, fc_nfields, 1, 3)));
	}
	fc_primary_sysid = pstrdup(PQgetvalue(fc_res, 0, 0));
	*fc_primary_tli = pg_strtoint32(PQgetvalue(fc_res, 0, 1));
	PQclear(fc_res);

	return fc_primary_sysid;
}

/*
 * libpq 的薄封装，以获取服务器版本。
 */
static int fc_libpqrcv_server_version(WalReceiverConn *fc_conn)
{
	return PQserverVersion(fc_conn->streamConn);
}

/*
 * 从给定的流选项开始流式传输 WAL 数据。
 *
 * 如果我们成功切换到 copy-both 模式，则返回 true。false
 * 意味着服务器接收到命令并成功执行，但没有切换到复制模式。
 * 这意味着在请求的时间线和起始点上没有 WAL，因为服务器切换到
 * 另一个时间线在请求的起始点之前或时。失败时，
 * 抛出 ERROR。
 */
static bool fc_libpqrcv_startstreaming(WalReceiverConn *fc_conn,
						const WalRcvStreamOptions *fc_options)
{
	StringInfoData fc_cmd;
	PGresult   *fc_res;

	Assert(fc_options->logical == fc_conn->logical);
	Assert(fc_options->slotname || !fc_options->logical);

	initStringInfo(&fc_cmd);

	/* 构建命令。 */
	appendStringInfoString(&fc_cmd, "START_REPLICATION");
	if (fc_options->slotname != NULL)
		appendStringInfo(&fc_cmd, " SLOT \"%s\"",
						 fc_options->slotname);

	if (fc_options->logical)
		appendStringInfoString(&fc_cmd, " LOGICAL");

	appendStringInfo(&fc_cmd, " %X/%X", LSN_FORMAT_ARGS(fc_options->startpoint));

	/*
	 * 额外选项取决于我们是进行逻辑复制还是物理复制。
	 */
	if (fc_options->logical)
	{
		char	   *fc_pubnames_str;
		List	   *fc_pubnames;
		char	   *fc_pubnames_literal;

		appendStringInfoString(&fc_cmd, " (");

		appendStringInfo(&fc_cmd, "proto_version '%u'",
						 fc_options->proto.logical.proto_version);

		if (fc_options->proto.logical.streaming &&
			PQserverVersion(fc_conn->streamConn) >= 140000)
			appendStringInfoString(&fc_cmd, ", streaming 'on'");

		if (fc_options->proto.logical.twophase &&
			PQserverVersion(fc_conn->streamConn) >= 150000)
			appendStringInfoString(&fc_cmd, ", two_phase 'on'");

		fc_pubnames = fc_options->proto.logical.publication_names;
		fc_pubnames_str = fc_stringlist_to_identifierstr(fc_conn->streamConn, fc_pubnames);
		if (!fc_pubnames_str)
			ereport(ERROR,
					(errcode(ERRCODE_OUT_OF_MEMORY),	/* 可能的猜测 */
					 errmsg("could not start WAL streaming: %s",
							pchomp(PQerrorMessage(fc_conn->streamConn)))));
		fc_pubnames_literal = PQescapeLiteral(fc_conn->streamConn, fc_pubnames_str,
										   strlen(fc_pubnames_str));
		if (!fc_pubnames_literal)
			ereport(ERROR,
					(errcode(ERRCODE_OUT_OF_MEMORY),	/* 可能的猜测 */
					 errmsg("could not start WAL streaming: %s",
							pchomp(PQerrorMessage(fc_conn->streamConn)))));
		appendStringInfo(&fc_cmd, ", publication_names %s", fc_pubnames_literal);
		PQfreemem(fc_pubnames_literal);
		pfree(fc_pubnames_str);

		if (fc_options->proto.logical.binary &&
			PQserverVersion(fc_conn->streamConn) >= 140000)
			appendStringInfoString(&fc_cmd, ", binary 'true'");
#ifdef FDDM
		if (fc_options->forward_sub_names != NULL)
		{
			char	   *fc_forward_subs_str;
			List	   *fc_sub_names;
			char	   *fc_sub_names_literal;

			fc_sub_names = fc_options->forward_sub_names;

			fc_forward_subs_str = fc_stringlist_to_identifierstr(fc_conn->streamConn, fc_sub_names);
			if (!fc_forward_subs_str)
				ereport(ERROR,
						(errcode(ERRCODE_OUT_OF_MEMORY),	/* 可能的猜测 */
						 errmsg("could not start WAL streaming: %s",
								pchomp(PQerrorMessage(fc_conn->streamConn)))));
			fc_sub_names_literal = PQescapeLiteral(fc_conn->streamConn, fc_forward_subs_str,
											   strlen(fc_forward_subs_str));
			if (!fc_sub_names_literal)
				ereport(ERROR,
						(errcode(ERRCODE_OUT_OF_MEMORY),	/* 可能的猜测 */
						 errmsg("could not start WAL streaming: %s",
								pchomp(PQerrorMessage(fc_conn->streamConn)))));
			appendStringInfo(&fc_cmd, ", forward_origin_nodenames %s", fc_sub_names_literal);
			PQfreemem(fc_sub_names_literal);
			pfree(fc_forward_subs_str);
		}
		
#endif
		appendStringInfoChar(&fc_cmd, ')');
	}
	else
		appendStringInfo(&fc_cmd, " TIMELINE %u",
						 fc_options->proto.physical.startpointTLI);

	/* 开始流式传输。 */
	fc_res = fc_libpqrcv_PQexec(fc_conn->streamConn, fc_cmd.data);
	pfree(fc_cmd.data);

	if (PQresultStatus(fc_res) == PGRES_COMMAND_OK)
	{
		PQclear(fc_res);
		return false;
	}
	else if (PQresultStatus(fc_res) != PGRES_COPY_BOTH)
	{
		PQclear(fc_res);
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("could not start WAL streaming: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));
	}
	PQclear(fc_res);
	return true;
}

/*
 * 停止流式传输 WAL 数据。 将下一个时间线的 ID 返回到 *next_tli，正如
 * 服务器报告的那样，或如果它没有报告，则为 0。
 */
static void fc_libpqrcv_endstreaming(WalReceiverConn *fc_conn, TimeLineID *fc_next_tli)
{
	PGresult   *fc_res;

	/*
	 * 发送 copy-end 消息。 正如在 libpqrcv_PQexec 中，这在理论上会阻塞，
	 * 但风险似乎很小。
	 */
	if (PQputCopyEnd(fc_conn->streamConn, NULL) <= 0 ||
		PQflush(fc_conn->streamConn))
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not send end-of-streaming message to primary: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));

	*fc_next_tli = 0;

	/*
	 * 在 COPY 完成后，我们应该接收到一个结果集，指示下一个时间线的 ID，
	 * 如果服务器关闭，则仅返回 CommandComplete。
	 *
	 * 如果我们尚未从后端接收 CopyDone，PGRES_COPY_OUT 也是可能的，
	 * 万一我们在流中途中止了复制。
	 */
	fc_res = fc_libpqrcv_PQgetResult(fc_conn->streamConn);
	if (PQresultStatus(fc_res) == PGRES_TUPLES_OK)
	{
		/*
		 * 读取下一个时间线的 ID。 服务器还发送时间线的
		 * 起始点，但它被忽略。
		 */
		if (PQnfields(fc_res) < 2 || PQntuples(fc_res) != 1)
			ereport(ERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("unexpected result set after end-of-streaming")));
		*fc_next_tli = pg_strtoint32(PQgetvalue(fc_res, 0, 0));
		PQclear(fc_res);

		/* 结果集应跟随 CommandComplete */
		fc_res = fc_libpqrcv_PQgetResult(fc_conn->streamConn);
	}
	else if (PQresultStatus(fc_res) == PGRES_COPY_OUT)
	{
		PQclear(fc_res);

		/* 结束复制 */
		if (PQendcopy(fc_conn->streamConn))
			ereport(ERROR,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("error while shutting down streaming COPY: %s",
							pchomp(PQerrorMessage(fc_conn->streamConn)))));

		/* CommandComplete 应该跟随 */
		fc_res = fc_libpqrcv_PQgetResult(fc_conn->streamConn);
	}

	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("error reading result of streaming command: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));
	PQclear(fc_res);

	/* 验证没有更多结果 */
	fc_res = fc_libpqrcv_PQgetResult(fc_conn->streamConn);
	if (fc_res != NULL)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("unexpected result after CommandComplete: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));
}

/*
 * 从主服务器获取 'tli' 的时间线历史文件。
 */
static void fc_libpqrcv_readtimelinehistoryfile(WalReceiverConn *fc_conn,
								 TimeLineID fc_tli, char **fc_filename,
								 char **fc_content, int *fc_len)
{
	PGresult   *fc_res;
	char		fc_cmd[64];

	Assert(!fc_conn->logical);

	/*
	 * 请求主服务器发送给定时间线的历史文件。
	 */
	snprintf(fc_cmd, sizeof(fc_cmd), "TIMELINE_HISTORY %u", fc_tli);
	fc_res = fc_libpqrcv_PQexec(fc_conn->streamConn, fc_cmd);
	if (PQresultStatus(fc_res) != PGRES_TUPLES_OK)
	{
		PQclear(fc_res);
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("could not receive timeline history file from "
						"the primary server: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));
	}
	if (PQnfields(fc_res) != 2 || PQntuples(fc_res) != 1)
	{
		int			fc_ntuples = PQntuples(fc_res);
		int			fc_nfields = PQnfields(fc_res);

		PQclear(fc_res);
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("invalid response from primary server"),
				 errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
						   fc_ntuples, fc_nfields)));
	}
	*fc_filename = pstrdup(PQgetvalue(fc_res, 0, 0));

	*fc_len = PQgetlength(fc_res, 0, 1);
	*fc_content = palloc(*fc_len);
	memcpy(*fc_content, PQgetvalue(fc_res, 0, 1), *fc_len);
	PQclear(fc_res);
}

/*
 * 发送查询并使用异步 libpq 函数和套接字就绪事件等待结果。
 *
 * 我们不能使用常规的阻塞 libpq 函数，如 PQexec()
 * 因为它们在某些平台（如 Windows）上无法被信号中断。
 *
 * 该函数的模型基于 libpq 中的 PQexec()，但仅实现
 * 在 walreceiver api 中使用的部分。
 *
 * 失败时可能返回 NULL，而不是错误结果。
 */
static PGresult * fc_libpqrcv_PQexec(PGconn *fc_streamConn, const char *fc_query)
{
	PGresult   *fc_lastResult = NULL;

	/*
	 * PQexec() 默默丢弃连接上的任何先前查询结果。
	 * 这个功能对于此函数并不是必需的，因为预计调用者
	 * （在所有情况下都是该库）会正确行为，我们不
	 * 必须与旧的 libpq 向后兼容。
	 */

	/*
	 * 提交查询。 由于我们不使用非阻塞模式，这在理论上可能
	 * 阻塞。 在实践中，由于我们不发送非常长的查询
	 * 字符串，因此风险似乎微不足道。
	 */
	if (!PQsendQuery(fc_streamConn, fc_query))
		return NULL;

	for (;;)
	{
		/* 等待并收集下一个 PGresult。 */
		PGresult   *fc_result;

		fc_result = fc_libpqrcv_PQgetResult(fc_streamConn);
		if (fc_result == NULL)
			break;				/* 查询已完成，或失败 */

		/*
		 * 模拟 PQexec() 的行为，当有许多结果时返回最后一个结果。
		 * 我们可以只返回最后的错误消息。
		 */
		PQclear(fc_lastResult);
		fc_lastResult = fc_result;

		if (PQresultStatus(fc_lastResult) == PGRES_COPY_IN ||
			PQresultStatus(fc_lastResult) == PGRES_COPY_OUT ||
			PQresultStatus(fc_lastResult) == PGRES_COPY_BOTH ||
			PQstatus(fc_streamConn) == CONNECTION_BAD)
			break;
	}

	return fc_lastResult;
}

/*
 * 执行 PQgetResult() 的等效操作，但要注意中断。
 */
static PGresult * fc_libpqrcv_PQgetResult(PGconn *fc_streamConn)
{
	/*
	 * 收集数据，直到 PQgetResult 准备好获取结果而不阻塞。
	 */
	while (PQisBusy(fc_streamConn))
	{
		int			fc_rc;

		/*
		 * 我们不需要将睡眠分解为更小的增量，
		 * 因为我们会被信号打断，并且可以在这里处理任何
		 * 中断。
		 */
		fc_rc = WaitLatchOrSocket(MyLatch,
							   WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
							   WL_LATCH_SET,
							   PQsocket(fc_streamConn),
							   0,
							   WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);

		/* 中断了吗？ */
		if (fc_rc & WL_LATCH_SET)
		{
			ResetLatch(MyLatch);
			ProcessWalRcvInterrupts();
		}

		/* 消耗来自套接字的所有可用数据 */
		if (PQconsumeInput(fc_streamConn) == 0)
		{
			/* 问题; 返回 NULL */
			return NULL;
		}
	}

	/* 现在我们可以收集并返回下一个 PGresult */
	return PQgetResult(fc_streamConn);
}

/*
 * 断开与主节点的连接（如果有的话）。
 */
static void fc_libpqrcv_disconnect(WalReceiverConn *fc_conn)
{
	PQfinish(fc_conn->streamConn);
	if (fc_conn->recvBuf != NULL)
		PQfreemem(fc_conn->recvBuf);
	pfree(fc_conn);
}

/*
 * 从 XLOG 流接收可用消息。
 *
 * 返回：
 *
 *	 如果接收到数据，则返回数据长度。*buffer 被设定为
 *	 指向包含接收到的消息的缓冲区。该缓冲区仅在
 *	 下一个 libpqrcv_* 调用之前有效。
 *
 *	 如果没有立即可用的数据，则返回 0，并且 *wait_fd 被设置为
 *	 一个可以在尝试再次之前等待的套接字描述符。
 *
 *	 如果服务器结束了 COPY，则返回 -1。
 *
 * 在错误时发出 ereport。
 */
static int fc_libpqrcv_receive(WalReceiverConn *fc_conn, char **fc_buffer,
				 pgsocket *fc_wait_fd)
{
	int			fc_rawlen;

	if (fc_conn->recvBuf != NULL)
		PQfreemem(fc_conn->recvBuf);
	fc_conn->recvBuf = NULL;

	/* 尝试接收CopyData消息 */
	fc_rawlen = PQgetCopyData(fc_conn->streamConn, &fc_conn->recvBuf, 1);
	if (fc_rawlen == 0)
	{
		/* 尝试消耗一些数据。 */
		if (PQconsumeInput(fc_conn->streamConn) == 0)
			ereport(ERROR,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("could not receive data from WAL stream: %s",
							pchomp(PQerrorMessage(fc_conn->streamConn)))));

		/* 既然我们已经消费了一些输入，再试一次 */
		fc_rawlen = PQgetCopyData(fc_conn->streamConn, &fc_conn->recvBuf, 1);
		if (fc_rawlen == 0)
		{
			/* 告诉调用者在我们的套接字准备好时重试。 */
			*fc_wait_fd = PQsocket(fc_conn->streamConn);
			return 0;
		}
	}
	if (fc_rawlen == -1)			/* 流结束或错误 */
	{
		PGresult   *fc_res;

		fc_res = fc_libpqrcv_PQgetResult(fc_conn->streamConn);
		if (PQresultStatus(fc_res) == PGRES_COMMAND_OK)
		{
			PQclear(fc_res);

			/* 验证没有更多结果。 */
			fc_res = fc_libpqrcv_PQgetResult(fc_conn->streamConn);
			if (fc_res != NULL)
			{
				PQclear(fc_res);

				/*
				 * 如果另一方有序地关闭了连接（否则
				 * 我们会看到一个错误，或 PGRES_COPY_IN）这里不报告错误，
				 * 而让调用者处理它。
				 */
				if (PQstatus(fc_conn->streamConn) == CONNECTION_BAD)
					return -1;

				ereport(ERROR,
						(errcode(ERRCODE_PROTOCOL_VIOLATION),
						 errmsg("unexpected result after CommandComplete: %s",
								PQerrorMessage(fc_conn->streamConn))));
			}

			return -1;
		}
		else if (PQresultStatus(fc_res) == PGRES_COPY_IN)
		{
			PQclear(fc_res);
			return -1;
		}
		else
		{
			PQclear(fc_res);
			ereport(ERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("could not receive data from WAL stream: %s",
							pchomp(PQerrorMessage(fc_conn->streamConn)))));
		}
	}
	if (fc_rawlen < -1)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("could not receive data from WAL stream: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));

	/* 将接收到的消息返回给调用者 */
	*fc_buffer = fc_conn->recvBuf;
	return fc_rawlen;
}

/*
 * 向 XLOG 流发送消息。
 *
 * 在错误时发出 ereport。
 */
static void fc_libpqrcv_send(WalReceiverConn *fc_conn, const char *fc_buffer, int fc_nbytes)
{
	if (PQputCopyData(fc_conn->streamConn, fc_buffer, fc_nbytes) <= 0 ||
		PQflush(fc_conn->streamConn))
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not send data to WAL stream: %s",
						pchomp(PQerrorMessage(fc_conn->streamConn)))));
}

/*
 * 创建新的复制槽。
 * 返回逻辑槽的导出快照的名称或物理槽的 NULL。
 */
static char * fc_libpqrcv_create_slot(WalReceiverConn *fc_conn, const char *fc_slotname,
#ifdef FDD
					 bool fc_temporary, bool fc_two_phase, bool fc_failover, CRSSnapshotAction fc_snapshot_action,
#else
					 bool temporary, bool two_phase, CRSSnapshotAction snapshot_action,
#endif
					 XLogRecPtr *fc_lsn)
{
	PGresult   *fc_res;
	StringInfoData fc_cmd;
	char	   *fc_snapshot;
	int			fc_use_new_options_syntax;

	fc_use_new_options_syntax = (PQserverVersion(fc_conn->streamConn) >= 150000);

	initStringInfo(&fc_cmd);

	appendStringInfo(&fc_cmd, "CREATE_REPLICATION_SLOT \"%s\"", fc_slotname);

	if (fc_temporary)
		appendStringInfoString(&fc_cmd, " TEMPORARY");

	if (fc_conn->logical)
	{
#ifdef FDDM
		if(MySubscription != NULL && strstr(MySubscription->name, "fmmr_") == MySubscription->name)
		{
			appendStringInfoString(&fc_cmd, " LOGICAL fddoutput ");
		}
		else
		{
			appendStringInfoString(&fc_cmd, " LOGICAL pgoutput ");
		}
#else
		appendStringInfoString(&fc_cmd, " LOGICAL pgoutput ");
#endif
		if (fc_use_new_options_syntax)
			appendStringInfoChar(&fc_cmd, '(');
		if (fc_two_phase)
		{
			appendStringInfoString(&fc_cmd, "TWO_PHASE");
			if (fc_use_new_options_syntax)
				appendStringInfoString(&fc_cmd, ", ");
			else
				appendStringInfoChar(&fc_cmd, ' ');
		}

		if (fc_use_new_options_syntax)
		{
			switch (fc_snapshot_action)
			{
				case CRS_EXPORT_SNAPSHOT:
					appendStringInfoString(&fc_cmd, "SNAPSHOT 'export'");
					break;
				case CRS_NOEXPORT_SNAPSHOT:
					appendStringInfoString(&fc_cmd, "SNAPSHOT 'nothing'");
					break;
				case CRS_USE_SNAPSHOT:
					appendStringInfoString(&fc_cmd, "SNAPSHOT 'use'");
					break;
			}
		}
		else
		{
			switch (fc_snapshot_action)
			{
				case CRS_EXPORT_SNAPSHOT:
					appendStringInfoString(&fc_cmd, "EXPORT_SNAPSHOT");
					break;
				case CRS_NOEXPORT_SNAPSHOT:
					appendStringInfoString(&fc_cmd, "NOEXPORT_SNAPSHOT");
					break;
				case CRS_USE_SNAPSHOT:
					appendStringInfoString(&fc_cmd, "USE_SNAPSHOT");
					break;
			}
		}

		if (fc_use_new_options_syntax)
			appendStringInfoChar(&fc_cmd, ')');
	}
	else
	{
		if (fc_use_new_options_syntax)
			appendStringInfoString(&fc_cmd, " PHYSICAL (RESERVE_WAL)");
		else
			appendStringInfoString(&fc_cmd, " PHYSICAL RESERVE_WAL");
	}

#ifdef FDD
	if (fc_failover)
	{
		appendStringInfoString(&fc_cmd, " FAILOVER");
	}
#endif

	fc_res = fc_libpqrcv_PQexec(fc_conn->streamConn, fc_cmd.data);
	pfree(fc_cmd.data);

	if (PQresultStatus(fc_res) != PGRES_TUPLES_OK)
	{
		PQclear(fc_res);
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("could not create replication slot \"%s\": %s",
						fc_slotname, pchomp(PQerrorMessage(fc_conn->streamConn)))));
	}

	if (fc_lsn)
		*fc_lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
												   CStringGetDatum(PQgetvalue(fc_res, 0, 1))));

	if (!PQgetisnull(fc_res, 0, 2))
		fc_snapshot = pstrdup(PQgetvalue(fc_res, 0, 2));
	else
		fc_snapshot = NULL;

	PQclear(fc_res);

	return fc_snapshot;
}

/*
 * 返回远程后端进程的 PID。
 */
static pid_t fc_libpqrcv_get_backend_pid(WalReceiverConn *fc_conn)
{
	return PQbackendPID(fc_conn->streamConn);
}

/*
 * 将元组查询结果转换为元组存储。
 */
static void fc_libpqrcv_processTuples(PGresult *fc_pgres, WalRcvExecResult *fc_walres,
					   const int fc_nRetTypes, const Oid *fc_retTypes)
{
	int			fc_tupn;
	int			fc_coln;
	int			fc_nfields = PQnfields(fc_pgres);
	HeapTuple	fc_tuple;
	AttInMetadata *fc_attinmeta;
	MemoryContext fc_rowcontext;
	MemoryContext fc_oldcontext;

	/* 确保我们得到了预期的字段数量。 */
	if (fc_nfields != fc_nRetTypes)
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("invalid query response"),
				 errdetail("Expected %d fields, got %d fields.",
						   fc_nRetTypes, fc_nfields)));

	fc_walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);

	/* 创建与预期结果相对应的元组描述符。 */
	fc_walres->tupledesc = CreateTemplateTupleDesc(fc_nRetTypes);
	for (fc_coln = 0; fc_coln < fc_nRetTypes; fc_coln++)
		TupleDescInitEntry(fc_walres->tupledesc, (AttrNumber) fc_coln + 1,
						   PQfname(fc_pgres, fc_coln), fc_retTypes[fc_coln], -1, 0);
	fc_attinmeta = TupleDescGetAttInMetadata(fc_walres->tupledesc);

	/* 如果没有返回元组，那么继续做其他事情没有意义。 */
	if (PQntuples(fc_pgres) == 0)
		return;

	/* 为本地分配创建临时上下文。 */
	fc_rowcontext = AllocSetContextCreate(CurrentMemoryContext,
									   "libpqrcv query result context",
									   ALLOCSET_DEFAULT_SIZES);

	/* 处理返回的行。 */
	for (fc_tupn = 0; fc_tupn < PQntuples(fc_pgres); fc_tupn++)
	{
		char	   *fc_cstrs[MaxTupleAttributeNumber];

		ProcessWalRcvInterrupts();

		/* 在临时上下文中进行分配。 */
		fc_oldcontext = MemoryContextSwitchTo(fc_rowcontext);

		/*
		 * 用以 null 结尾的字符串填充 cstrs，表示列值。
		 */
		for (fc_coln = 0; fc_coln < fc_nfields; fc_coln++)
		{
			if (PQgetisnull(fc_pgres, fc_tupn, fc_coln))
				fc_cstrs[fc_coln] = NULL;
			else
				fc_cstrs[fc_coln] = PQgetvalue(fc_pgres, fc_tupn, fc_coln);
		}

		/* 将行转换为元组，并将其添加到元组存储中 */
		fc_tuple = BuildTupleFromCStrings(fc_attinmeta, fc_cstrs);
		tuplestore_puttuple(fc_walres->tuplestore, fc_tuple);

		/* 清理 */
		MemoryContextSwitchTo(fc_oldcontext);
		MemoryContextReset(fc_rowcontext);
	}

	MemoryContextDelete(fc_rowcontext);
}

/*
 * 发送通用查询（和命令）的公共接口。
 *
 * 只能从连接到数据库的进程中调用。
 */
static WalRcvExecResult *
fc_libpqrcv_exec(WalReceiverConn *fc_conn, const char *fc_query,
			  const int fc_nRetTypes, const Oid *fc_retTypes)
{
	PGresult   *fc_pgres = NULL;
	WalRcvExecResult *fc_walres = palloc0(sizeof(WalRcvExecResult));
	char	   *fc_diag_sqlstate;

	if (MyDatabaseId == InvalidOid)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("the query interface requires a database connection")));

	fc_pgres = fc_libpqrcv_PQexec(fc_conn->streamConn, fc_query);

	switch (PQresultStatus(fc_pgres))
	{
		case PGRES_SINGLE_TUPLE:
		case PGRES_TUPLES_OK:
			fc_walres->status = WALRCV_OK_TUPLES;
			fc_libpqrcv_processTuples(fc_pgres, fc_walres, fc_nRetTypes, fc_retTypes);
			break;

		case PGRES_COPY_IN:
			fc_walres->status = WALRCV_OK_COPY_IN;
			break;

		case PGRES_COPY_OUT:
			fc_walres->status = WALRCV_OK_COPY_OUT;
			break;

		case PGRES_COPY_BOTH:
			fc_walres->status = WALRCV_OK_COPY_BOTH;
			break;

		case PGRES_COMMAND_OK:
			fc_walres->status = WALRCV_OK_COMMAND;
			break;

			/* 空查询被视为错误。 */
		case PGRES_EMPTY_QUERY:
			fc_walres->status = WALRCV_ERROR;
			fc_walres->err = _("empty query");
			break;

		case PGRES_PIPELINE_SYNC:
		case PGRES_PIPELINE_ABORTED:
			fc_walres->status = WALRCV_ERROR;
			fc_walres->err = _("unexpected pipeline mode");
			break;

		case PGRES_NONFATAL_ERROR:
		case PGRES_FATAL_ERROR:
		case PGRES_BAD_RESPONSE:
			fc_walres->status = WALRCV_ERROR;
			fc_walres->err = pchomp(PQerrorMessage(fc_conn->streamConn));
			fc_diag_sqlstate = PQresultErrorField(fc_pgres, PG_DIAG_SQLSTATE);
			if (fc_diag_sqlstate)
				fc_walres->sqlstate = MAKE_SQLSTATE(fc_diag_sqlstate[0],
												 fc_diag_sqlstate[1],
												 fc_diag_sqlstate[2],
												 fc_diag_sqlstate[3],
												 fc_diag_sqlstate[4]);
			break;
	}

	PQclear(fc_pgres);

	return fc_walres;
}

/*
 * 给定一个字符串列表，将其作为单个逗号分隔的
 * 字符串返回，根据需要引用标识符。
 *
 * 这本质上是 SplitIdentifierString 的反向操作。
 *
 * 调用者应该释放结果。
 */
static char * fc_stringlist_to_identifierstr(PGconn *fc_conn, List *fc_strings)
{
	ListCell   *fc_lc;
	StringInfoData fc_res;
	bool		fc_first = true;

	initStringInfo(&fc_res);

	foreach(fc_lc, fc_strings)
	{
		char	   *fc_val = strVal(lfirst(fc_lc));
		char	   *fc_val_escaped;

		if (fc_first)
			fc_first = false;
		else
			appendStringInfoChar(&fc_res, ',');

		fc_val_escaped = PQescapeIdentifier(fc_conn, fc_val, strlen(fc_val));
		if (!fc_val_escaped)
		{
			free(fc_res.data);
			return NULL;
		}
		appendStringInfoString(&fc_res, fc_val_escaped);
		PQfreemem(fc_val_escaped);
	}

	return fc_res.data;
}
